\section{Model}\label{sec:model}

Compute-aggregate is a two-phase class of problems (see Figure~\ref{fig:highlevel}). The first computation phase
happens only at the leaf nodes. At each such node, data $z$ is
contained in memory prior to computation, and computation applies some
function $f$ such that $x = f(z)$, where $x$ is data ready for aggregation. Once
the computation phase is completed the aggregation overlay is responsible for
aggregating the partial results from all leaves into a single result. This is
done by applying a function $g$ which takes $x_{1} \ldots x_{i}$ and outputs the
aggregate $x^{1..i}$, i.e., $x^{1..i}= g(\overline{x})$. 
Table~\ref{tab:properties} defines more precisely the required traits of an
aggregation function which allows for configurable overlays. Note that we say
results are equivalent ($\equiv$) rather than equal.
For example, if you want the top-$k$ results and there are $k+1$ results with
the same score, different aggregation orders may result in different %$k$
results, but each is correct.

%, and works for any number of inputs up to the number of leaves
Each node applies the function to all of its inputs and forwards its results. As long as the function
is cumulative commutative, and associative, any overlay network that has exactly
one path from each leaf to the root of the aggregation tree gives a correct output.

%\begin{table}[b]
%  \centering
%  \footnotesize
%  \setlength{\tabcolsep}{.4em}
%\vspace{2mm}
%  \caption{Mathematical definitions of requirements for $g\left(
%  \overline{x}\right)$.  $\overline{x}^{\prime}$
%  and $\overline{x}^{\prime\prime}$ are defined in the associative property.}
%  \label{tab:properties}
%  \begin{tabular}{p{.22\columnwidth}|p{.73\columnwidth}}
%  Property & Definition\\ \hline
%  Associative & $g\left(\overline{x}\right) \equiv$
%  $g\left(g\left(\overline{x}^{\prime}\right),
%  \overline{x}^{\prime\prime}\right)$,
%  $\overline{x} = \overline{x}^{\prime} \cup \overline{x}^{\prime\prime}$, 
%  $\overline{x}^{\prime} \cap \overline{x}^{\prime\prime} = \emptyset$\\
%  Cumulative & $g\left( g\left( \overline{x}^{\prime}\right), g\left(
%  \overline{x}^{\prime\prime}\right)\right) \equiv$
%  $g\left(\overline{x}\right)$ \\
%  Commutative & $g\left(\overline{x}^{\prime}, \overline{x}^{\prime\prime}\right)
%  \equiv$ $g\left(\overline{x}^{\prime\prime}, \overline{x}^{\prime}\right)$
%  \end{tabular}
%\end{table}

\begin{table}[b]
  \centering
  \footnotesize
  \setlength{\tabcolsep}{.4em}
\vspace{2mm}
  \caption{Mathematical definitions of requirements for $g\left(
  \overline{x}\right)$.  
%$\overline{x}^{\prime}$ and $\overline{x}^{\prime\prime}$ are defined in the associative property.
}
  \label{tab:properties}
  \begin{tabular}{p{.22\columnwidth}|p{.73\columnwidth}}
  Property & Definition\\ \hline
  Cumulative & $g\left( g\left( \overline{x}\right), g\left(
  \overline{x}^{\prime}\right)\right) \equiv$
  $g\left(\overline{x},\overline{x}^{\prime}\right)$ \\
  Commutative & $g\left(\overline{x}^{\prime}, \overline{x}\right)
  \equiv$ $g\left(\overline{x}, \overline{x}^{\prime}\right) $\\
  Associative & $ g\left(g\left(\overline{x}\right),
  \overline{x}^{\prime}\right) \equiv g\left(\overline{x},\overline{x}^{\prime}\right)
  $\\
  \end{tabular}
\end{table}


With everything in memory the time to aggregate at each node is
mostly dependent upon the size of its combined input. The most efficient system
is one with no waiting at any nodes that have work to do, so we assume a
balanced and full tree; sibling aggregation nodes begin and finish at roughly
the same time, with a bit of a buffer provided by the nature of network
communication. This does require some level of homogeneity, which is not
unreasonable to expect in cloud services. If any assumptions are broken, the
optimality guarantee breaks, but the correctness of the output remains.

\begin{figure}
\resizebox{\linewidth}{!}{%
\begin{tikzpicture}[-,auto,,
  lines/.style={transparent}]
  \node [circle, draw] (n7) at (6,4) {};
  \node [circle, draw] (n6) at (7,3) {};
  \node [circle, draw] (n5) at (5,3) {};
  \node [circle, draw] (n4) at (8,2) {};
  \node [circle, draw] (n3) at (4,2) {};
  \node [rectangle, draw] (n01) at (3,1) {};
  \node [rectangle, draw] (n02) at (5,1) {};
  \node [rectangle, draw] (n03) at (7,1) {};
  \node [rectangle, draw] (n04) at (9,1) {};
  
  \node [] (agglabel) at (0,3) {Aggregation Overlay};
  \node [] (complabel) at (0,1) {Computation at Local Nodes};

  \path[every node/.style={font=\sffamily\small}]
    (n7)edge [] node[left]{} (n6)
        edge [] node[left]{} (n5)
    (n6)edge [line width=1.25pt, line cap=round, dash pattern=on 0pt off 6\pgflinewidth ] node[left]{} (n4)
    (n5)edge [line width=1.25pt, line cap=round, dash pattern=on 0pt off 6\pgflinewidth] node[left]{} (n3)
    (n3)edge [] node[left]{} (n01)
        edge [] node[left]{} (n02)
    (n4)edge [] node[left]{} (n03)
        edge [] node[left]{} (n04);

\path (n5) -- (n6) node [midway] {$\cdots$};
\path (n3) -- (n4) node [midway] {$\cdots$};
\path (n01) -- (n02) node [midway] {$\cdots$};
\path (n02) -- (n03) node [midway] {$\cdots$};
\path (n03) -- (n04) node [midway] {$\cdots$};
\draw [-,decorate,decoration=snake] (0,1.5) -- (9.5,1.5);
\end{tikzpicture}
}
\caption{Computation and aggregation phases.}
\label{fig:highlevel}
\end{figure}

Our objective is to minimize the total latency of the system. Since the
aggregation phase is separate from the local computation time, we can optimize them
separately. The aggregation overlay simply aggregates the outputs from all the
computation nodes, so the number of leaf nodes for the tree
defining the aggregation overlay is given by the number of local
nodes involved in the computation phase. The functions to aggregate multiple
inputs into a single input are also given by the problem, so the only variable
left to change is the \emph{fanout}. Figure~\ref{fig:fanoutvariants} shows three
equivalent trees with different fanouts.

The aggregation time at a level --- composed of the time to receive input
from the level just beneath it and the time to create the output for the level ---
depends on the size of the input, some set of partial results $\overline{x}$. We
use $g^{c}\left(\overline{x}\right)$ to denote the \emph{time required} by
$g\left( \overline{x}\right)$ to aggregate input of size $\left|
\overline{x}\right|$, including communication time. Aggregation on the same level of the overlay happens in parallel, so only the time of a single branch is
modeled.

Fanout and tree height are inversely related. Increased height
increases amount of work done in parallel at the lower levels, but it also might
increase the total amount of work to be done when results have to filter up through more
levels. We analyze traits of the aggregation function to determine if the time
saved by the parallelism offsets the time required by extra levels.

\begin{figure}
\centering
\subfigure[Fanout = 2]{
\resizebox{.45\linewidth}{.10\linewidth}{%
\begin{tikzpicture}[-,shorten >=1pt,auto,
  thin,main node/.style={circle,draw,font=\sffamily\Large\bfseries}]
  [scale=.25,auto=right,every node/.style={circle}]
  \node (n31) at (16,5) {};
  \node (n30) at (24,4) {};
  \node (n29) at (8,4) {};
  \node (n28) at (28,3) {};
  \node (n27) at (20,3) {};
  \node (n26) at (12,3) {};
  \node (n25) at (4,3) {};
  \node (n24) at (30,2) {};
  \node (n23) at (26,2) {};
  \node (n22) at (22,2) {};
  \node (n21) at (18,2) {};
  \node (n20) at (14,2) {};
  \node (n19) at (10,2) {};
  \node (n18) at (6,2) {};
  \node (n17) at (2,2) {};
  \node (n16) at (31,1) {};
  \node (n15) at (29,1) {};
  \node (n14) at (27,1) {};
  \node (n13) at (25,1) {};
  \node (n12) at (23,1) {};
  \node (n11) at (21,1) {};
  \node (n10) at (19,1) {};
  \node (n9) at (17,1) {};
  \node (n8) at (15,1) {};
  \node (n7) at (13,1) {};
  \node (n6) at (11,1) {};
  \node (n5) at (9,1) {};
  \node (n4) at (7,1) {};
  \node (n3) at (5,1) {};
  \node (n2) at (3,1) {};
  \node (n1) at (1,1) {};
        
    \path[every node/.style={font=\sffamily\small}]
    (n31)edge [] node[left]{} (n30)
        edge [] node[left]{} (n29)
    (n30)edge [] node[left]{} (n28)
        edge [] node[left]{} (n27)
    (n29)edge [] node[left]{} (n26)
        edge [] node[left]{} (n25)
    (n28)edge [] node[left]{} (n24)
        edge [] node[left]{} (n23)
    (n27)edge [] node[left]{} (n22)
        edge [] node[left]{} (n21)
    (n26)edge [] node[left]{} (n20)
        edge [] node[left]{} (n19)
    (n25)edge [] node[left]{} (n18)
        edge [] node[left]{} (n17)
    (n24)edge [] node[left]{} (n16)
        edge [] node[left]{} (n15)
    (n23)edge [] node[left]{} (n14)
        edge [] node[left]{} (n13)
    (n22)edge [] node[left]{} (n12)
        edge [] node[left]{} (n11)
    (n21)edge [] node[left]{} (n10)
        edge [] node[left]{} (n9)
    (n20)edge [] node[left]{} (n8)
        edge [] node[left]{} (n7)
    (n19)edge [] node[left]{} (n6)
        edge [] node[left]{} (n5)
    (n18)edge [] node[left]{} (n4)
        edge [] node[left]{} (n3)
    (n17)edge [] node[left]{} (n2)
        edge [] node[left]{} (n1);
\end{tikzpicture} } }
\subfigure[Fanout = 4]{
\resizebox{.45\linewidth}{.10\linewidth}{%
\begin{tikzpicture}[-,shorten >=1pt,auto,
  ultra thin,main node/.style={circle,draw,font=\sffamily\Large\bfseries}]
  [scale=.40,auto=right,every node/.style={circle}]
  \node (n21) at (16,3) {};
  \node (n20) at (28,2) {};
  \node (n19) at (20,2) {};
  \node (n18) at (12,2) {};
  \node (n17) at (4,2) {};
  \node (n16) at (31,1) {};
  \node (n15) at (29,1) {};
  \node (n14) at (27,1) {};
  \node (n13) at (25,1) {};
  \node (n12) at (23,1) {};
  \node (n11) at (21,1) {};
  \node (n10) at (19,1) {};
  \node (n9) at (17,1) {};
  \node (n8) at (15,1) {};
  \node (n7) at (13,1) {};
  \node (n6) at (11,1) {};
  \node (n5) at (9,1) {};
  \node (n4) at (7,1) {};
  \node (n3) at (5,1) {};
  \node (n2) at (3,1) {};
  \node (n1) at (1,1) {};
        
    \path[every node/.style={font=\sffamily\small}]
    (n21)edge [] node[left]{} (n20)
        edge [] node[left]{} (n19)
		edge [] node[left]{} (n18)
        edge [] node[left]{} (n17)
    (n20)edge [] node[left]{} (n16)
        edge [] node[left]{} (n15)
		edge [] node[left]{} (n14)
        edge [] node[left]{} (n13)
    (n19)edge [] node[left]{} (n12)
        edge [] node[left]{} (n11)
		edge [] node[left]{} (n10)
        edge [] node[left]{} (n9)
    (n18)edge [] node[left]{} (n8)
        edge [] node[left]{} (n7)
		edge [] node[left]{} (n6)
        edge [] node[left]{} (n5)
    (n17)edge [] node[left]{} (n4)
        edge [] node[left]{} (n3)
		edge [] node[left]{} (n2)
        edge [] node[left]{} (n1);
\end{tikzpicture} } }\\
\subfigure[Fanout = 16]{
\resizebox{.45\linewidth}{.10\linewidth}{%
\begin{tikzpicture}[-,shorten >=1pt,auto,
  ultra thin,main node/.style={circle,draw,font=\sffamily\Large\bfseries}]
  [scale=.60,auto=right,every node/.style={circle}]
  \node (n17) at (16,2.75) {};
  \node (n16) at (31,.5) {};
  \node (n15) at (29,.5) {};
  \node (n14) at (27,.5) {};
  \node (n13) at (25,.5) {};
  \node (n12) at (23,.5) {};
  \node (n11) at (21,.5) {};
  \node (n10) at (19,.5) {};
  \node (n9) at (17,.5) {};
  \node (n8) at (15,.5) {};
  \node (n7) at (13,.5) {};
  \node (n6) at (11,.5) {};
  \node (n5) at (9,.5) {};
  \node (n4) at (7,.5) {};
  \node (n3) at (5,.5) {};
  \node (n2) at (3,.5) {};
  \node (n1) at (1,.5) {};
        
    \path[every node/.style={font=\sffamily\small}]
    (n17)edge [] node[left]{} (n16)
        edge [] node[left]{} (n15)
		edge [] node[left]{} (n14)
        edge [] node[left]{} (n13)
		edge [] node[left]{} (n12)
        edge [] node[left]{} (n11)
		edge [] node[left]{} (n10)
        edge [] node[left]{} (n9)
		edge [] node[left]{} (n8)
        edge [] node[left]{} (n7)
		edge [] node[left]{} (n6)
        edge [] node[left]{} (n5)
		edge [] node[left]{} (n4)
        edge [] node[left]{} (n3)
		edge [] node[left]{} (n2)
        edge [] node[left]{} (n1);
\end{tikzpicture}
}
}
\caption{Three aggregation overlays with 16 leaves.}
\label{fig:fanoutvariants}
\end{figure}

Table~\ref{tab:heuristicsummary} summarizes our optimal heuristics for
choosing the fanout of the aggregation overlay tree and notes the cases for
which they are proved.
(Proofs can be found at \cite{loom}.) Two variables in the table represent characteristics of the aggregation.
$g^{c}\left( \overline{x}\right)$ is one. The other is the ratio of the size of
the final aggregate output to the size of the output of a single computation
node, denoted  $y_{0}$. $n$ is simply the number of leaves.


\begin{table}[t]
  \centering
  \footnotesize
  \setlength{\tabcolsep}{.3em}
\vspace{4.5mm}

  \caption{Heuristic values for fanout. \checkmark\ means provably optimal, *~indicates near-optimal result.}
  \label{tab:heuristicsummary}
  \begin{tabular}{>{\centering}p{.16\columnwidth}|>{\centering\arraybackslash}p{.41\columnwidth}|>{\centering\arraybackslash}p{.10\columnwidth}|>{\centering\arraybackslash}p{.10\columnwidth}|>{\centering\arraybackslash}p{.12\columnwidth}}
  {{${y_{0}}$ }} & 
  {{Optimal fanout}} &
  {{Sublin. ${g^{c}\left( x\right)}$}} & 
  {{Linear ${g^{c}\left( x\right)}$}} &
  {{Superlin. ${g^{c}\left( x\right)}$ }}\\ \hline
  %$y \downarrow $ & & & \\ \hline
  $y_{0} < 1 $	& 2 & ~ & {\checkmark}   &
  {\checkmark}   \\
  $y_{0} = 1 $ & $e$ & ~ & {\checkmark}   &
  {\checkmark} $\mathbf{*}$ \\
  %\hline
%   $1 < y_{0} < n $ & $min\left( n, \left( 1 -
% \frac{\log y_{0}}{\log n}\right)^{-\frac{\log n}{\log y_{0}}}
% \right)$ & ~
  $1 < y_{0} < n $ & $min\left( n, \left( 1 - \log_{n} y_{0}
\right)^{-\log_{y_{0}} n}
\right)$ & ~
  & {\checkmark}  & ~ \\
  %\hline
  $n \leq y_{0} $ & $n$ & {\checkmark}  &
  {\checkmark}  & {\checkmark}   
  \end{tabular}
\end{table}


Optimal fanouts remain uproven when the degree
of sublinearity or superlinearity is necessary for meaningful analysis.
Communication time is linear with respect to the input size, so
there is always linear component to $g^{c}\left(\overline{x}\right)$. Thus
it makes sense to use the heuristics from the linear cases on the sublinear cases.
In practice we also use the heuristics from the linear cases on the superlinear
cases as they are the best available analyses.

$y_{0}$ is a very relevant variable to aggregation. As seen in
Table~\ref{tab:heuristicsummary}, it is the primary factor for deciding the
optimal fanout of the aggregation overlay. To give an idea what types of
problems fall into each category we present Table~\ref{tab:y0examples}. It shows
some aggregation functions which fall into some of the different regions for
$y_{0}$.

\begin{table}[h]
  \centering
  \footnotesize
\setlength{\extrarowheight}{-1.8ex}
\vspace{2mm}
  \caption{Typical aggregation functions $g$ grouped by $y_{0}$.}
  \label{tab:y0examples}
  \begin{tabular}{>{\centering}p{.10\columnwidth}|p{.78\columnwidth}}
    {$y_{0}$} & {Common problems}
    \\%[.2ex]
    \hline ~&~\\%[-2.0ex]
    $y_{0} < 1$ & The average MapReduce job at Google~\cite{MapReduce}, the
    average ``aggregate'' jobs at Facebook and Yahoo~\cite{YahooFBStat}
    \\[.2ex]~&~\\%[-2.0ex]
    $y_{0} = 1$ & Min, max, average, top-$k$ match, word count with a fixed
    dictionary, multiplication of square matrices\\%[.2ex]
~&~\\%[-2.0ex]
    $y_{0} > 1$ & sort, concatenate, word count with
    disjoint dictionaries
  \end{tabular}
\end{table}
